package com.eshipenet.shipnet.dp_decoder

import java.util.Properties

import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer

object EdgeChannelDataPacketSinkProvider {

  def create(env: StreamExecutionEnvironment): SinkFunction[EdgeChannelDataPacket] = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "kafka-server:9092")
    properties.setProperty("group.id", "EdgeChannelDataPacketDecodedProducer")
    val kafkaSink = new FlinkKafkaProducer[EdgeChannelDataPacket](
      "EdgeChannelDataPacketDecoded2",
      new EdgeChannelDataPacketSerializer,
      properties,
      FlinkKafkaProducer.Semantic.NONE
    )
    kafkaSink.setWriteTimestampToKafka(true)
    kafkaSink
  }

}
